package io.kimmking.rpcfx.register.center.zookeeper;

import com.alibaba.fastjson.JSON;
import io.kimmking.rpcfx.api.Router;
import io.kimmking.rpcfx.proxy.CustomInterceptor;
import io.kimmking.rpcfx.register.center.api.Discovery;
import io.kimmking.rpcfx.register.center.api.DiscoveryInfo;
import io.kimmking.rpcfx.register.center.api.ProviderInfo;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.ArrayList;
import java.util.List;

/**
 * @author lfy
 */
public class ZookeeperDiscovery implements Discovery {


    private CuratorFramework client;
    private final String zookeeperServer;
    private final String namespace;

    public ZookeeperDiscovery(String server, String namespace) {
        this.zookeeperServer = server;
        this.namespace = namespace;
        init();
    }
    private void init() {
        // start zk client
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
         client = CuratorFrameworkFactory.builder().connectString(zookeeperServer)
                .namespace(namespace).retryPolicy(retryPolicy).build();
        client.start();

    }
    @Override
    public List<ProviderInfo> discovery(DiscoveryInfo discoveryInfo, CustomInterceptor handler, Router router) {
        try {
            String path = "/" +discoveryInfo.getServiceName();
            subscribe(path,handler,router,discoveryInfo);
            return discoveryProviderUrls(handler, path,router,discoveryInfo);
        } catch (Exception e) {
            e.printStackTrace();
        }

        return null;
    }

    @Override
    public List<ProviderInfo> discovery(DiscoveryInfo discoveryInfo, Router router) {
        String path = "/" +discoveryInfo.getServiceName();
        try {
            return discoveryProviderUrls(null, path,router,discoveryInfo);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    private  List<ProviderInfo> discoveryProviderUrls(CustomInterceptor handler, String path ,Router router,DiscoveryInfo discoveryInfo) throws Exception {
        List<ProviderInfo> result = new ArrayList<>();
        List<String> list = client.getChildren().forPath(path);
        for (String childPath : list) {
            findOne(result,path+"/"+childPath);
        }
        List<ProviderInfo >providers = router.route(result,discoveryInfo);
        if(providers==null ||providers.size() == 0){
            System.out.println("没有找到服务!");
        }
        if(handler!=null){
            handler.setProviders(providers);
        }
        return result;
    }

    /**
     * 注册监听节点
     */
    public void subscribe(String path, CustomInterceptor handler, Router router,DiscoveryInfo discoveryInfo){
        try {

            PathChildrenCache cache = new PathChildrenCache(client, path, true);
            cache.start();
            PathChildrenCacheListener cacheListener = (client1, event) -> {
                System.err.println("Event Type ：" + event.getType());
                //重新设置url
                discoveryProviderUrls(handler, path,router,discoveryInfo);
//                switch (event.getType()){
//                    case CHILD_ADDED:
//                        break;
//                    case CHILD_REMOVED:
//                        break;
//                    case CHILD_UPDATED:
//                        break;
//
//                }

            };
            cache.getListenable().addListener(cacheListener);

        } catch (Exception e) {
            e.printStackTrace();
        }

    }
    private void findOne(List<ProviderInfo> result, String childPath) throws Exception {
        byte[] bytes = client.getData().forPath(childPath);
        ProviderInfo providerInfo = JSON.parseObject(new String(bytes), ProviderInfo.class);
        result.add(providerInfo);

    }
}
